package com.atguigu.flink.chapter02_DataStreamAPI.source;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * Created by Smexy on 2022/10/22
 *
 *  UnsupportedFileSystemSchemeException:
 *              Hadoop is not in the classpath/dependencies.
 *              flink本身不带hadoop，需要hadoop的依赖自己加进来
 */
public class Demo4_HDFSFile
{
    public static void main(String[] args) {
        
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        /*
            filePath可以是一个目录，也可以是一个具体文件的路径
         */
        DataStreamSource<String> source = env.readTextFile("hdfs://hadoop102:9820/input");


        source
            .flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>()
            {

                @Override
                public void flatMap(String line, Collector<Tuple2<String, Integer>> out) throws Exception {
                    String[] words = line.split(" ");
                    for (String word : words) {
                        out.collect(Tuple2.of(word, 1));
                    }
                }
            })
            .keyBy(new KeySelector<Tuple2<String, Integer>, String>()
            {
                @Override
                public String getKey(Tuple2<String, Integer> value) throws Exception {
                    return value.f0;
                }
            })
            .sum(1)
            .print();
        
        try {
                    env.execute();
                } catch (Exception e) {
                    e.printStackTrace();
                }
    }
}
